package com.atguigu.loginfail_detect

import java.util

import org.apache.flink.cep.{PatternFlatSelectFunction, PatternSelectFunction}
import org.apache.flink.cep.scala.{CEP, PatternStream}
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

/**
 * 2秒内连续2次登陆
 * 恶意登录监控 使用CEP
 *
 * Project: UserBehaviorAnalysis
 * Package: com.atguigu.loginfail_detect
 * Version: 1.0
 *
 * Created by  WangJX  on 2019/12/13 19:57
 */
object LoginFail {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val path: String = LoginFail.getClass.getClassLoader.getResource("LoginLog.csv").getPath

    val value: DataStream[LoginEvent] = env.readTextFile(path)
      .map(
        data => {
          val dataArrays: Array[String] = data.split(",")
          LoginEvent(dataArrays(0).toLong, dataArrays(1), dataArrays(2), dataArrays(3).toLong)
        }
      )
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[LoginEvent](Time.seconds(5)) {
        override def extractTimestamp(element: LoginEvent): Long = element.eventTime * 1000L
      })
      .keyBy(_.userId)


    val filterPattern: Pattern[LoginEvent, LoginEvent] = Pattern.begin[LoginEvent]("start").where(_.eventType == "fail")
      .next("next").where(_.eventType == "fail")
      .within(Time.seconds(2))



    val patternData: PatternStream[LoginEvent] = CEP.pattern(value, filterPattern)



    //select 模式选择函数可以生成一个结果元素。
    val ouput: DataStream[Warning] = patternData.select(new MySelect())

    //flatSelect 模式平面选择函数可以生成任意数量的结果元素。
//    patternData.flatSelect(new MyFlatSelect())

    ouput.print("output").setParallelism(1)

    env.execute("LoginFail job")
  }

}

case class LoginEvent(
                       userId: Long,
                       ip: String,
                       eventType: String,
                       eventTime: Long
                     )

case class Warning(
                  firstTime: Long,
                  LastTime: Long,
                  msg: String
                  )

class MySelect() extends PatternSelectFunction[LoginEvent, Warning] {
  override def select(map: util.Map[String, util.List[LoginEvent]]): Warning = {
    val start = map.get("start").iterator()
    val last = map.get("next").iterator()

    Warning(start.next().eventTime, last.next().eventTime, "times more 2 !")
  }
}

//class MyFlatSelect() extends PatternFlatSelectFunction[LoginEvent, Warning]{
//  override def flatSelect(map: util.Map[String, util.List[LoginEvent]], collector: Collector[Warning]): Unit = {
//
//    collector.collect()
//  }
//}